import os
import csv
import time
import json
from kafka import KafkaProducer
from dateutil import parser


def read_csv_file(file_path):
    """读取CSV文件并返回数据列表"""
    data = []
    try:
        with open(file_path, 'r', encoding='utf-8-sig') as file:
            reader = csv.DictReader(file)
            for row in reader:
                # 处理缺失值
                for key in row:
                    if row[key].strip() == '':
                        row[key] = None

                # 转换日期时间格式
                for field in ['order_created', 'order_paid']:
                    if row.get(field):
                        try:
                            # 尝试解析日期时间
                            dt = parser.parse(row[field])
                            row[field] = dt.isoformat()
                        except ValueError:
                            print(f"警告: 无法解析日期时间 {field}: {row[field]}")
                            row[field] = None

                # 转换数值类型
                for field in ['total_amount', 'amount_paid', 'refund_amount']:
                    if row.get(field) is not None:
                        try:
                            row[field] = float(row[field])
                        except ValueError:
                            print(f"警告: 无法将 {field} 转换为数值: {row[field]}")
                            row[field] = None

                data.append(row)

        print(f"成功读取 {len(data)} 条记录")
        return data
    except Exception as e:
        print(f"读取CSV文件时出错: {e}")
        return []


def send_to_kafka(data, bootstrap_servers, topic, interval=1.0):
    """将数据发送到Kafka主题"""
    try:
        # 创建Kafka生产者实例
        producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v,ensure_ascii=False).encode('utf-8'),
            retries=5  # 重试次数
        )

        total_records = len(data)
        print(f"准备发送 {total_records} 条记录到Kafka主题: {topic}")

        for i, record in enumerate(data):
            # 发送消息到Kafka
            producer.send(topic, record)
            print(f"已发送记录 {i + 1}/{total_records}: 订单编号={record['order_id']}")

            # 按照指定间隔等待
            time.sleep(interval)

        # 确保所有消息都被发送
        producer.flush()
        producer.close()
        print(f"所有 {total_records} 条记录已成功发送到Kafka")

    except Exception as e:
        print(f"发送数据到Kafka时出错: {e}")


if __name__ == "__main__":
    # 配置参数
    config = {
        'csv_file': 'tmall_order_report.csv',  # CSV文件路径
        'bootstrap_servers': ['10.90.100.101:9092'],  # Kafka服务器地址
        'topic': 'orders_topic',  # Kafka主题
        'send_interval': 0.1  # 发送间隔(秒)
    }


    # 读取CSV数据
    csv_data = read_csv_file("tmall_order_report.csv")

    if csv_data:
        # 发送到Kafka
        send_to_kafka(
            data=csv_data,
            bootstrap_servers=config['bootstrap_servers'],
            topic=config['topic'],
            interval=config['send_interval']
        )